<?php

namespace app\services;

use app\payment\PayFactory;
use plugin\admin\app\model\Notice;
use plugin\admin\app\model\Order;
use plugin\admin\app\model\TransferJob;
use plugin\admin\app\model\User;
use plugin\admin\app\model\UserDownloadLog;
use plugin\admin\app\model\UserLevel;
use plugin\admin\app\model\UserPropertyLog;
use support\Cache;
use support\coze\Factory;
use support\coze\OauthClient;
use support\JRTools;
use support\Log;
use support\PdfToImg;
use Webman\RedisQueue\Redis;

class TransferService
{
    public static function addJob($data)
    {
        $job = new TransferJob();
        $job->user_id = get_user_id();
        $job->params = json_encode($data);
        $job->created_time = date('Y-m-d H:i:s');
        $job->status = TransferJob::PENDING;
        $job->need_pay = 1;
        $job->file_name = $data['file_name'];
        $job->is_img = $data['is_img'] ?? 0;
        $job->save();

        Redis::send('coze-transfer', [
            'id' => $job->id,
        ]);

        self::sendJobMessage($job->id, '任务已添加，正在排队中，请稍等...');

        return $job->id;
    }

    public static function addDontNeedTransferJob($data)
    {
        $channelCount = $data['channel_count'] ?? 1;

        $attributes = [];

        foreach (range(1, $channelCount) as $index) {
            $attributes[] = [
                'channels' => [[$index]],
                'enName' => '',
                'name' => '点击选择通道',
                'subAttributes' => [],
            ];
        }

        $job = new TransferJob();
        $job->user_id = get_user_id();
        $job->params = json_encode($data);
        $job->created_time = date('Y-m-d H:i:s');
        $job->status = TransferJob::EXECUTE_SUCCESS;
        $job->need_pay = 1;
        $job->file_name = $data['file_name'] ?? '自定义文件';
        $job->is_img = $data['is_img'] ?? 0;
        $job->coze_execute_result = json_encode([
            "attributes" => $attributes
        ], JSON_UNESCAPED_UNICODE);
        $job->transfer_file_name = $data['transfer_file_name'] ?? '';
        $job->save();

        return $job->id;
    }

    public static function getExecutor($jobId)
    {
        $job = TransferJob::query()->where('id', $jobId)->first();

        if ($job->is_img) {
            $params = json_decode($job['params'], true);
            if (is_array($params['file_path'])) {
                self::execute2($jobId);
            } else {
                self::execute($jobId);
            }
        } else {

            $extension = pathinfo($job->file_name, PATHINFO_EXTENSION);

            if ($extension == 'pdf') {
                self::execute3($jobId);
            } else {
                self::execute($jobId);
            }
        }
    }

    public static function execute3($jobId)
    {
        ini_set('memory_limit', -1);
        $job = TransferJob::query()->where('id', $jobId)->first();

        $workFlowId = self::getWorkFlowId('png');

        $job->coze_workflow_id = $workFlowId;

        self::sendJobMessage($job->id, '正在初始化AI引擎...', 5);

        $app = OauthClient::getInstance(getenv('COZE_APP_ID'), config_path() . '/coze/private_key.pem', getenv('COZE_PUBLIC_KEY_ID'));

        self::sendJobMessage($job->id, 'AI引擎初始化成功...', 10);

        $params = json_decode($job['params'], true);

        $ocrWorkFlowId = '7507105902136459276';

        self::sendJobMessage($job->id, '正在从pdf中提取图片...', 15);

        try {
            $extractPath = public_path() . '/uploads/pdfimg/' . random_int(10000, 99999) . '/';
            $filePath = str_replace(getenv('APP_URL'), '', $params['file_path']);

            $pdf = new PdfToImg(public_path() . $filePath, $extractPath);
            $images = $pdf->extractWithPdftoPpm();

            if ($images) {
                self::sendJobMessage($job->id, '图片提取成功...', 20);
            } else {
                throw new \Exception('非图片型pdf，暂不支持转换');
            }

            self::sendJobMessage($job->id, '正在上传图片到AI进行分析...', 25);

            try {
                $totalCozeFileIds = [];
                $fileSize = formatFileSize(public_path() . $filePath);
                foreach ($images as $image) {
                    $res = $app->uploadFile($image);
                    $totalCozeFileIds[] = ['file_id' => $res['id'], 'image' => $image];
                }

            } catch (\Exception $e) {
                self::sendJobMessage($job->id, '图片上传失败.原因为' . $e->getMessage(), 90);
                throw $e;
            }

            $fileIds = [];

            $cozeFileIds = array_chunk($totalCozeFileIds, 5);

            print_r($cozeFileIds);

            self::sendJobMessage($job->id, '图片上传成功，AI引擎正在进行OCR识别...', 30);

            foreach ($cozeFileIds as $key => $value) {
                $input = [];
                foreach ($value as $item) {
                    $input[] = ['file_id' => $item['file_id']];
                }
                self::sendJobMessage($job->id, '正在OCR识别第' . ($key + 1) . '批图片...', 30);

                print_r($input);
                $res = $app->runWorkFlow($ocrWorkFlowId, [
                    'input' => $input
                ]);

                if ($res['code'] != 0) {
                    continue;
                }
                $outputs = json_decode($res['data'], true);

                foreach ($outputs['output'] as $k => $v) {
                    if ($v['output'] == 1) {
                        $fileIds[] = ['file_id' => $value[$k]['file_id']];
                    }
                }
            }

            if (!$fileIds) {
                throw new \Exception('未识别到有效的通道图片');
            }

            $job->coze_file_id = json_encode($fileIds);
            $job->file_size = $fileSize;

            $fileIdValues = array_column($fileIds, 'file_id');

            try {
                self::sendJobMessage($job->id, '图片识别成功，AI引擎正在进行转换...', 35);

                $channelImages = [];
                foreach ($totalCozeFileIds as $cozeFileId) {
                    if (in_array($cozeFileId['file_id'], $fileIdValues)) {
                        $channelImages[] = $cozeFileId['image'];
                    }
                }

                $mergedImagePath = public_path() . '/uploads/pdfimg/' . $job->id . '.png';
                mergeImages($channelImages, 'vertical', $mergedImagePath);

                $mergedImageRes = $app->uploadFile($mergedImagePath);

                print_r($mergedImageRes);

                $res = $app->asyncRunWorkFlow($workFlowId, [
                    'input' => [
                        ['file_id' => $mergedImageRes['id']]
                    ]
                ]);

                $job->coze_debug_url = $res['debug_url'];

                self::sendJobMessage($job->id, '转换期间该页面可以关闭...', 40);
                self::sendJobMessage($job->id, '稍后会在消息中心通知您...', 45);

            } catch (\Exception $e) {
                self::sendJobMessage($job->id, 'AI引擎转换失败，原因为' . $e->getMessage(), 90);
                throw $e;
            }

            $job->coze_execute_id = $res['execute_id'];
            $job->save();


        } catch (\Exception $e) {
            echo $e->getMessage() . $e->getTraceAsString();
            $job->status = TransferJob::EXECUTE_FAILED;
            $job->save();
            self::sendJobMessage($job->id, '任务执行失败结束', 100, 400);
        }

    }

    public static function execute2($jobId)
    {
        $job = TransferJob::query()->where('id', $jobId)->first();

        $workFlowId = self::getWorkFlowId('png');

        $job->coze_workflow_id = $workFlowId;

        self::sendJobMessage($job->id, '正在初始化AI引擎...', 5);

        $app = OauthClient::getInstance(getenv('COZE_APP_ID'), config_path() . '/coze/private_key.pem', getenv('COZE_PUBLIC_KEY_ID'));

        self::sendJobMessage($job->id, 'AI引擎初始化成功...', 10);

        $params = json_decode($job['params'], true);


        try {
            self::sendJobMessage($job->id, '正在上传文件到AI分析...', 15);

            try {
                $cozeFileIds = [];
                $fileSize = 0;
                foreach ($params['file_path'] as $filePath) {
                    $filePath = str_replace(getenv('APP_URL'), '', $filePath);
                    $fileSize = formatFileSize(public_path() . $filePath);

                    if (strpos($fileSize, 'MB') !== false) {
                        self::sendJobMessage($job->id, '文件较大，整个转换过程可能长达10分钟以上', 18);
                    }
                    $res = $app->uploadFile(public_path() . $filePath);
                    $cozeFileIds[] = $res['id'];
                }

            } catch (\Exception $e) {
                self::sendJobMessage($job->id, '文件上传失败.原因为' . $e->getMessage(), 90);
                throw $e;
            }

            $job->coze_file_id = json_encode($cozeFileIds);
            $job->file_size = $fileSize;

            try {
                self::sendJobMessage($job->id, '文件上传成功，AI引擎正在进行转换...', 20);

                $res = $app->asyncRunWorkFlow($workFlowId, [
                    'input' => [
                        ['file_id' => $cozeFileIds[0]],
                        ['file_id' => $cozeFileIds[1]]
                    ]
                ]);

                $job->coze_debug_url = $res['debug_url'];

                self::sendJobMessage($job->id, '转换期间该页面可以关闭...', 25);
                self::sendJobMessage($job->id, '稍后会在消息中心通知您...', 30);

            } catch (\Exception $e) {
                self::sendJobMessage($job->id, 'AI引擎转换失败，原因为' . $e->getMessage(), 90);
                throw $e;
            }

            $job->coze_execute_id = $res['execute_id'];
            $job->save();


        } catch (\Exception $e) {
            echo $e->getMessage() . $e->getTraceAsString();
            $job->status = TransferJob::EXECUTE_FAILED;
            $job->save();
            self::sendJobMessage($job->id, '任务执行失败结束', 100, 400);
        }

    }

    public static function execute($jobId)
    {
        $job = TransferJob::query()->where('id', $jobId)->first();

        $workFlowId = self::getWorkFlowId(pathinfo($job->file_name, PATHINFO_EXTENSION));

        $job->coze_workflow_id = $workFlowId;

        self::sendJobMessage($job->id, '正在初始化AI引擎...', 5);

        $app = OauthClient::getInstance(getenv('COZE_APP_ID'), config_path() . '/coze/private_key.pem', getenv('COZE_PUBLIC_KEY_ID'));

        self::sendJobMessage($job->id, 'AI引擎初始化成功...', 10);

        $params = json_decode($job['params'], true);


        try {
            self::sendJobMessage($job->id, '正在上传文件到AI分析...', 15);

            try {
                $filePath = str_replace(getenv('APP_URL'), '', $params['file_path']);
                $fileSize = formatFileSize(public_path() . $filePath);

                if (strpos($fileSize, 'MB') !== false) {
                    self::sendJobMessage($job->id, '文件较大，整个转换过程可能长达10分钟以上', 18);
                }
                $res = $app->uploadFile(public_path() . $filePath);
            } catch (\Exception $e) {
                self::sendJobMessage($job->id, '文件上传失败.原因为' . $e->getMessage(), 90);
                throw $e;
            }

            $job->coze_file_id = $res['id'];
            $job->file_size = $fileSize;

            try {
                self::sendJobMessage($job->id, '文件上传成功，AI引擎正在进行转换...', 20);

                $res = $app->asyncRunWorkFlow($workFlowId, [
                    'input' => [
                        'file_id' => $res['id']
                    ]
                ]);

                $job->coze_debug_url = $res['debug_url'];

                self::sendJobMessage($job->id, '转换期间该页面可以关闭...', 25);
                self::sendJobMessage($job->id, '稍后会在消息中心通知您...', 30);

            } catch (\Exception $e) {
                self::sendJobMessage($job->id, 'AI引擎转换失败，原因为' . $e->getMessage(), 90);
                throw $e;
            }

            $job->coze_execute_id = $res['execute_id'];
            $job->save();


        } catch (\Exception $e) {
            echo $e->getMessage() . $e->getTraceAsString();
            $job->status = TransferJob::EXECUTE_FAILED;
            $job->save();
            self::sendJobMessage($job->id, '任务执行失败结束', 100, 400);
        }

    }

    public static function getJob($jobId)
    {
        $res = TransferJob::query()->where('id', $jobId)->where('user_id', get_user_id())->first();

        if ($res['download_url']) {
            $downloadUrl = $res['download_url'];

            if (!$res->transfer_file_name && $res['status'] == 1) {
                $res->transfer_file_name = getFilenameFromUrl($res['download_url']);
                $res->save();
            }

            $res['download_url'] = self::getDownloadUrl($res);
        }

        $res['coze_execute_result'] = $res['after_edit_result'] ? json_decode($res['after_edit_result'], true) : json_decode($res['coze_execute_result'], true);

        return $res;
    }

    public static function getDownloadUrl($res)
    {
        if (is_array($res)) {
            $res = TransferJob::where('id', $res['id'])->first();
        }

        $downloadUrl = $res['download_url'];

        if (empty($downloadUrl)) {
            return '';
        }

        if ($res['need_pay'] && !$res['pay_time']) {
            $userInfo = User::where('id', get_user_id())->first();

            if ($userInfo['download_count'] > 0) {
                $userInfo['download_count'] = $userInfo['download_count'] - 1;
                $userInfo->save();

                $res['pay_time'] = date('Y-m-d H:i:s');
            }

            $res->save();
        }


        if (!$res['need_pay'] || $res['pay_time']) {
            return getenv('APP_URL') . '/api/download-url?redirect_url=' . $downloadUrl . '&token=' . request()->header('token');
        }



        return '';
    }

    public static function getDownloadUrl2($res)
    {
        $downloadUrl = $res['download_url'];

        if (!$res['need_pay'] || $res['pay_time']) {
            return getenv('APP_URL') . '/api/download-url2?redirect_url=' . $downloadUrl . '&token=' . request()->header('token');
        }

        return '';
    }

    public static function downloadRecordList($page, $length = 10, $keyword = '')
    {
        $offset = ($page - 1) * $length;
        $userId = get_user_id();

        // Get all download logs with pagination first
        $downloadLogsQuery = \plugin\admin\app\model\UserDownloadLog::query()
            ->where('user_id', $userId)
            ->orderBy('created_time', 'desc')
            ->offset($offset)
            ->limit($length);

        // Clone the query to get total count (without pagination)
        $countQuery = clone $downloadLogsQuery;
        $total = $countQuery->count();

        // Get the logs for this page
        $downloadLogs = $downloadLogsQuery->get()->toArray();

        if (empty($downloadLogs)) {
            return [
                'total' => 0,
                'page' => $page,
                'length' => $length,
                'list' => [],
            ];
        }

        // Organize download logs by type to handle job_id collisions
        $jobLogs = [];
        $productLogs = [];

        foreach ($downloadLogs as $log) {
            if ($log['type'] == UserDownloadLog::TYPE_JOB) {
                $jobLogs[] = $log;
            } else {
                $productLogs[] = $log;
            }
        }

        // Extract job IDs by type
        $jobIds = array_column($jobLogs, 'job_id');
        $productIds = array_column($productLogs, 'job_id');

        // Get job data for TYPE_JOB records
        $jobData = [];
        if (!empty($jobIds)) {
            $jobsQuery = \plugin\admin\app\model\TransferJob::query()
                ->whereIn('id', $jobIds)
                ->select(['id', 'transfer_file_name', 'file_name', 'download_url', 'need_pay', 'pay_time']);

            // Apply keyword filter to jobs if provided
            if ($keyword) {
                $jobsQuery->where(function ($query) use ($keyword) {
                    $query->where('transfer_file_name', 'like', "%{$keyword}%")
                        ->orWhere('file_name', 'like', "%{$keyword}%");
                });
            }

            // Use keyBy to create an associative array with job_id as key
            $jobData = $jobsQuery->get()->keyBy('id')->toArray();
        }

        // Get product data for non-TYPE_JOB records
        $productData = [];
        if (!empty($productIds)) {
            $productsQuery = \plugin\admin\app\model\UserCompanyProduct::query()
                ->whereIn('id', $productIds)
                ->select(['id', 'file_name', 'download_url']);

            // Apply keyword filter to products if provided
            if ($keyword) {
                $productsQuery->where('file_name', 'like', "%{$keyword}%");
            }

            // Use keyBy to create an associative array with id as key
            $productData = $productsQuery->get()->keyBy('id')->toArray();
        }

        // Combine the results and prepare the final records list
        $records = [];
        $filteredCount = 0;

        // Process job logs
        foreach ($jobLogs as $log) {
            $jobId = $log['job_id'];

            // Skip if the job doesn't exist or was filtered out by keyword
            if (!isset($jobData[$jobId])) {
                continue;
            }

            $job = $jobData[$jobId];

            $records[] = [
                'transfer_file_name' => $job['transfer_file_name'],
                'file_name' => $job['file_name'],
                'type' => $log['type'],
                'created_time' => $log['created_time'],
                'download_url' => self::getDownloadUrl($job)
            ];

            $filteredCount++;
        }

        // Process product logs
        foreach ($productLogs as $log) {
            $productId = $log['job_id'];

            // Skip if the product doesn't exist or was filtered out by keyword
            if (!isset($productData[$productId])) {
                continue;
            }

            $product = $productData[$productId];

            $downloadRecord = [
                'need_pay' => 1,
                'pay_time' => true,
                'download_url' => $product['download_url'] ?? '/api/product/download/' . $productId
            ];

            $records[] = [
                'transfer_file_name' => $product['file_name'], // Products don't have transfer_file_name
                'type' => $log['type'],
                'created_time' => $log['created_time'],
                'download_url' => self::getDownloadUrl2($downloadRecord)
            ];

            $filteredCount++;
        }

        // Sort records by created_time in descending order to maintain correct order
        usort($records, function ($a, $b) {
            return strtotime($b['created_time']) - strtotime($a['created_time']);
        });

        // If we applied keyword filtering, adjust total count
        if ($keyword) {
            $total = $filteredCount;
        }

        return [
            'total' => $total,
            'page' => $page,
            'length' => $length,
            'list' => $records,
        ];
    }

    public static function createOrder($jobId, $payWay, $payWayType = 'h5')
    {
        $user = User::where('id', get_user_id())->first();
        // Generate a unique order number
        $orderNo = OrderService::generateOrderNo();

        // Create the order
        $orderData = [
            'order_no' => $orderNo,
            'user_id' => get_user_id(),
            'price' => ConfigService::get('need_pay_point'), // Example price for upgrading membership
            'type' => Order::TYPE_WORK,
            'pay_way' => $payWay,
            'pay_way_type' => $payWayType,
            'status' => Order::STATUS_WAITING_PAY,
            'created_time' => date('Y-m-d H:i:s'),
        ];
        Order::insert($orderData);

        TransferJob::where('id', $jobId)->update(['order_no' => $orderNo]);

        // Prepare the response
        $response = ['order_no' => $orderNo];

        // If payment method is WeChat, generate WeChat payment parameters
        if ($payWay === Order::PAY_TYPE_WECHAT) {
            $paymentParams = PayFactory::create($payWay)->setData([
                'pay_way_type' => $payWayType,
                'order_no' => $orderNo,
                'total_amount' => $orderData['price'],
            ])->pay();

            $response['payment_params'] = $paymentParams;
        } elseif ($payWay === Order::PAY_TYPE_BALANCE) {
            PayFactory::create($payWay)->setData([
                'order_no' => $orderNo,
                'total_amount' => $orderData['price'],
                'source' => UserPropertyLog::PAY_ORDER
            ])->pay();
        }

        if ($payWay == Order::PAY_TYPE_BALANCE) {
            self::paySuccess($orderNo);
        }

        return $response;
    }

    public static function paySuccess($orderNo)
    {
        $order = Order::where('order_no', $orderNo)->first();

        $dateTime = date('Y-m-d H:i:s');

        $order->status = Order::STATUS_FINISH;
        $order->pay_time = $dateTime;
        $order->save();

        $job = TransferJob::where('order_no', $orderNo)->first();

        $job->pay_time = $dateTime;
        $job->save();

        Notice::add($order['user_id'], '下载支付', sprintf("您已成功支付¥%.2f，可下载《%s》", $order['price'], $job['transfer_file_name']), Notice::TYPE_PAYMENT_DOWNLOAD, $job['id']);
    }

    public static function queryExecuteResult()
    {
        // Limit to 5 jobs per execution for better performance
        $jobs = TransferJob::where('status', TransferJob::PENDING)
            ->whereNotNull('coze_execute_id')
            ->limit(5)
            ->get();

        if ($jobs->isEmpty()) {
            return;
        }

        // Create API client once to avoid multiple initializations
        $app = OauthClient::getInstance(getenv('COZE_APP_ID'), config_path() . '/coze/private_key.pem', getenv('COZE_PUBLIC_KEY_ID'));

        $app->setDebug(true);

        foreach ($jobs as $job) {
            try {
                // Track processing status in Redis to avoid redundant checks
                $cacheKey = "job_processing_{$job->id}";

                // Query job status from Coze API
                $res = $app->queryRunWorkFlow($job->coze_workflow_id, $job->coze_execute_id);

                Log::info("Processing job {$job->id} with response: " . json_encode($res));

                if ($res['code'] != 0) {
                    // API error - mark as failed
                    $job->status = TransferJob::EXECUTE_FAILED;
                    $job->save();
                    continue;
                }

                // Check if execution is complete
                if ($res['data'][0]['execute_status'] !== 'Success') {
                    self::sendJobMessage($job->id, '数据正在转换中，耗时较长，该页面可随时关闭...', 35);
                    continue;
                }

                // Process successful execution
                $job->status = TransferJob::EXECUTE_SUCCESS;
                $extension = pathinfo($job['file_name'], PATHINFO_EXTENSION);

                // Extract output based on file type
                try {
                    if ($job->is_img == 0 && $extension != 'pdf') {
                        $output = $res['data'][0]['output'];
                        $output = json_decode($output, true);
                        $output = json_decode($output['Output'], true);

                        if (isset($output['output'])) {
                            if (empty($output['output'])) {
                                throw new \Exception('该文档暂不支持转换');
                            }

                            $output = json_decode($output['output'], true);

                        } else {
                            $output = json_decode($output['output1'], true);
                        }
                    } else {
                        $output = $res['data'][0]['output'];
                        // Remove debug print_r to improve performance
                        $output = json_decode($output, true);
                        $output = json_decode($output['Output'], true);
                        $output = json_decode($output['text'], true);
                    }

                    $job->coze_execute_result = json_encode($output, JSON_UNESCAPED_UNICODE);
                    self::sendJobMessage($job->id, 'AI引擎转换成功', 100, 200, Notice::TYPE_EDIT);
                    $job->save();

                    // Process the file using JRTools
                    // JRTools::transfer($job->id);

                } catch (\Exception $e) {
                    $job->status = TransferJob::EXECUTE_FAILED;
                    $job->fail_reason = $e->getMessage() . $e->getTraceAsString();
                    $job->save();
                    self::sendJobMessage($job->id, '处理任务结果时出错: ' . $e->getMessage(), 100, 400);
                }

            } catch (\Exception $e) {
                // Handle any unexpected exceptions
                try {
                    $job->status = TransferJob::EXECUTE_FAILED;
                    $job->fail_reason = $e->getMessage() . $e->getTraceAsString();
                    $job->save();
                    self::sendJobMessage($job->id, '任务处理发生异常: ' . $e->getMessage(), 100, 400);
                } catch (\Exception $inner) {
                    // Log error if we can't even update the job status
                    error_log("Failed to process job {$job->id}: " . $e->getMessage() . " and failed to update status: " . $inner->getMessage());
                }
            }
        }
    }

    public static function getWorkFlowId($fileExtension)
    {
        switch ($fileExtension) {
            case 'doc':
            case 'docx':
                return '7501541332348698660';
            case 'png':
            case 'jpg':
            case 'pdf':
                return '7513099505215471655';

        }
    }

    public static function sendJobMessage($jobId, $message, $process = 0, $code = 201, $messageType = Notice::TYPE_TRANSFER)
    {
        $job = TransferJob::query()->where('id', $jobId)->first();

        if ($job->messages) {
            $messages = json_decode($job->messages, true);
        } else {
            $messages = [];
        }

        $data = [
            'code' => $code,
            'process' => $process,
            'message' => $message,
            'time' => date('Y-m-d H:i:s'),
        ];

        $messages[] = $data;

        $job->messages = json_encode($messages, JSON_UNESCAPED_UNICODE);
        $job->save();

        if ($process == 100) {
            if ($code == 200) {
                if ($messageType == Notice::TYPE_EDIT) {
                    $desc = sprintf("文档《%s》转换成功，点击二次编辑", $job['file_name']);
                } else {
                    $desc = sprintf("文档《%s》转换成功，点击下载", $job['file_name']);
                }
                Notice::add($job['user_id'], '文档转换', $desc, $messageType, $job['id']);
            } else {
                Notice::add($job['user_id'], '文档转换', sprintf("文档《%s》转换失败", $job['file_name']), $messageType, $job['id']);
            }
        }


        $redisKey = 'transfer-message-' . $jobId;

        // Push the message to the Redis queue
        \support\Redis::rpush($redisKey, json_encode($data));
    }

    public static function getJobMessage($jobId)
    {

        $redisKey = 'transfer-message-' . $jobId;

        // Pop the message from the Redis queue
        $message = \support\Redis::lpop($redisKey);

        return $message ? json_decode($message, true) : [
            'code' => 201,
            'process' => '',
            'message' => '',
            'time' => date('Y-m-d H:i:s')
        ];
    }

    /**
     * 获取当前登录用户的job列表，支持分页
     * @param int $page
     * @param int $perPage
     * @return array
     */
    public static function getJobList($page = 1, $perPage = 10)
    {
        $query = \plugin\admin\app\model\TransferJob::select([
            'id',
            'file_name',
            'transfer_file_name',
            'status',
            'order_no',
            'fail_reason',
            'need_pay',
            'params',
            'pay_time',
            'created_time',
        ])->where('user_id', get_user_id())->where('download_url', '!=', '');

        $total = $query->count();

        $jobs = $query->offset(($page - 1) * $perPage)
            ->limit($perPage)
            ->orderBy('id', 'desc')
            ->get()
            ->toArray();

        // 提取订单号
        $orderNos = array_column($jobs, 'order_no');
        $orders = [];
        if (!empty($orderNos)) {
            $orders = \plugin\admin\app\model\Order::select(['order_no', 'price', 'pay_way', 'pay_time', 'status'])
                ->whereIn('order_no', $orderNos)
                ->get()
                ->keyBy('order_no')
                ->toArray();
        }

        // 合并订单信息
        foreach ($jobs as &$job) {
            $order = $orders[$job['order_no']] ?? null;
            $job['order'] = $order;
            $job['status_text'] = \plugin\admin\app\model\TransferJob::getStatusText($job['status'] ?? 0);
            $job['download_url'] = self::getDownloadUrl($job);
        }

        return [
            'total' => $total,
            'page' => $page,
            'length' => $perPage,
            'data' => $jobs,
        ];
    }
}